#pragma once
#include <Message.pb.h>
#include <fmtlog/fmtlog.h>
#include <cstddef>
#include <unordered_map>
#include <unordered_set>
#include <mutex>
#include <memory>
#include <list>
#include <string>
#include <filesystem>
#include <fstream>
#include <Common/FileHelper.hpp>
#include <Common/UUIDHelper.hpp>
namespace MY_MQ
{
    using MessagePtr = std::shared_ptr<MQ_Protobuf::Message>;
    const std::string DATAFILE_SUBFIX = ".mqd";
    const std::string TMPFILE_SUBFIX = ".mqd.tmp";
    class MessageHelper
    {

    public:
        MessageHelper(std::string MsgFilepath)
        {
            // 用filesystem检测是否存在目录，不存在则创建目录以及文件
            if (!std::filesystem::exists(MsgFilepath))
            {
                std::filesystem::create_directories(MsgFilepath);
            }
            _DataFilepath = MsgFilepath + DATAFILE_SUBFIX;
            _TmpMsgFilepath = MsgFilepath + TMPFILE_SUBFIX;
            CreateMsgFile();
        }
        bool CreateMsgFile()
        {
            if (std::filesystem::exists(_DataFilepath))
            {

                return true;
            }
            std::filesystem::create_directories(std::filesystem::path(_DataFilepath).parent_path());
            std::ofstream ofs(_DataFilepath, std::ios::binary);
            if (false == ofs.is_open())
            {
                loge("{}文件打开失败", _DataFilepath);
                ofs.close();
                return false;
            }
            ofs.close();
            return true;
        }
        void RemoveMessageFile()
        {
            std::filesystem::remove(_DataFilepath);
            std::filesystem::remove(_TmpMsgFilepath);
        }
        bool remove(MessagePtr &msg)
        {
            // 1.将消息中有效标志位修改 为0
            msg->mutable_payload()->set_valid("0");
            // 2.msg序列化
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                logw("You can't modify the DataFile infomation,because the new infomation's length is not the same as before !!");
                return false;
            }

            // 3.序列化后的消息,写入数据在文件中
            MY_MQ::Filehelper file(_DataFilepath);
            if (false == file.writeBinary(body.c_str(), msg->offset(), body.size()))
            {
                loge("Failed to write data to queue data file !");
                return false;
            }
            logd("Delete the durable Messgae Success!");
            return true;
        }
        bool insert(MessagePtr msg)
        {
            std::cout << msg->payload().SerializeAsString() << std::endl;
            return insert(_DataFilepath, msg);
        }

        std::list<MessagePtr> GarbageCollector()
        {
            std::list<MessagePtr> res;
            if (false == load(res))
            {
                loge("load List error");
                return res;
            }
            logd("垃圾文件回收，得到的有效消息数量{}", res.size());
            Filehelper::create_file(_TmpMsgFilepath);
            for (auto &msg : res)
            {
                logd("write data to TMpFile,{}", msg->payload().body());
                if (false == insert(_TmpMsgFilepath, msg))
                {
                    loge("Failed to write to TmpFile");
                    return res;
                }
            }
            logd("回收垃圾后，临时文件数据写入完毕，临时文件大小：{} Byte", Filehelper(_TmpMsgFilepath).size());
            if (false == std::filesystem::remove(_DataFilepath))
            {
                logw("Failed to delete data Source File");
                return res;
            }
            if (false == Filehelper(_TmpMsgFilepath).rename(_DataFilepath))
            {
                logw("Failed to rename TmpFile to Data File");
                return res;
            }
            return res;
        }

    private:
        std::string _DataFilepath;
        std::string _TmpMsgFilepath;
        // load the data from file,storage format: 4 byte for data length | data
        bool load(std::list<MessagePtr> &result)
        {
            Filehelper dataFH(_DataFilepath);
            size_t offset = 0, msgsize;
            size_t fsize = dataFH.size();

            while (offset < fsize)
            {
                if (false == dataFH.read((char *)&msgsize, offset, sizeof(size_t)))
                {
                    loge("Failed to read the length of data !");
                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msgsize, 0);
                if (false == dataFH.read(&msg_body[0], offset, msgsize))
                {
                    loge("Failed To read messageData From File");
                    return false;
                }
                offset += msgsize;
                MessagePtr msgp = std::make_shared<MQ_Protobuf::Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                if (msgp->payload().valid() == "0")
                {
                    logw("load Invalid infomation");
                    continue;
                }
                result.push_back(msgp);
            }
            return true;
        }

        // 数据添加在文件末尾
        bool insert(const std::string &filename, MessagePtr msg)
        {
            // 序列化消息
            std::string body = msg->payload().SerializeAsString();
            MY_MQ::Filehelper fileHelper(filename);
            size_t filesize = fileHelper.size();
            size_t msg_size = body.size();

            // 写入过程，1. 先写入4字节数据长度 2.再写入指定数据长度
            if (false == fileHelper.writeBinary((char *)&msg_size, filesize, sizeof(size_t)))
            {
                loge("Failed to write the fileSize!");
                return false;
            }

            if (false == fileHelper.writeBinary(body.c_str(), filesize + sizeof(size_t), msg_size))
            {
                loge("Failed to write the Data to the File :{}!", filename);
                return false;
            }
            // 3.update the info in fact
            msg->set_offset(filesize + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }
    };
    class MessageQueue
    {
    public:
        using self_ptr = std::shared_ptr<MessageQueue>;
        MessageQueue(const std::string &DataFilePath)
            :  _DataFilePath(DataFilePath),_helper(DataFilePath), _valid_count(0), _total_count(0)
        {
        }
        bool recovery()
        {
            std::lock_guard lk(_mutex);
            _ToBePushed_msgList = _helper.GarbageCollector();
            for (auto &msg : _ToBePushed_msgList)
            {
                _durable_msgs[msg->payload().properties().id()] = msg;
            }
            _valid_count = _total_count = _ToBePushed_msgList.size();
            return true;
        }
        bool insert(const MQ_Protobuf::BasicProperties &bp, const std::string &body, bool queue_is_durable)
        {
            MessagePtr msg = std::make_shared<MQ_Protobuf::Message>();
            msg->mutable_payload()->set_body(body);
            MQ_Protobuf::DeliveryMode mode = queue_is_durable ? bp.delivery_mode() : MQ_Protobuf::DeliveryMode::UNDURABLE;
            msg->mutable_payload()->mutable_properties()->set_id(bp.id());
            msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            msg->mutable_payload()->mutable_properties()->set_routing_key(bp.routing_key());
            return InsertWithLock(msg);
        }
        bool insert(const std::string &body, bool queue_is_durable)
        {
            MessagePtr msg = std::make_shared<MQ_Protobuf::Message>();
            msg->mutable_payload()->set_body(body);
            MQ_Protobuf::DeliveryMode mode = queue_is_durable ? MQ_Protobuf::DURABLE : MQ_Protobuf::DeliveryMode::UNDURABLE;
            msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuidWithShuffledMAC());
            msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            msg->mutable_payload()->mutable_properties()->set_routing_key("");
            return InsertWithLock(std::move(msg));
        }
        MessagePtr front()
        {
            std::lock_guard lk(_mutex);
            if (_ToBePushed_msgList.empty())
            {
                return nullptr;
            }
            MessagePtr msg = _ToBePushed_msgList.front();
            _ToBePushed_msgList.pop_front();
            _waitConfirm_msgs[msg->payload().properties().id()] = msg;
            return msg;
        }

        bool remove(const std::string &msg_id)
        {
            std::lock_guard lk(_mutex);
            auto it = _waitConfirm_msgs.find(msg_id);
            if (_waitConfirm_msgs.end() == it)
            {
                // 说明已经没了
                return true;
            }
            if (it->second->payload().properties().delivery_mode() == MQ_Protobuf::DURABLE)
            {
                _helper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                GarbageCollector();
            }
            _waitConfirm_msgs.erase(msg_id);
            return true;
        }
        inline size_t getToBePushedMsg_count()
        {
            std::lock_guard lk(_mutex);
            return _ToBePushed_msgList.size();
        }
        inline size_t total_count()
        {
            std::lock_guard lk(_mutex);
            return _total_count;
        }
        inline size_t waitConfirm_count()
        {
            std::lock_guard lk(_mutex);
            return _waitConfirm_msgs.size();
        }
        inline size_t durable_count()
        {
            std::lock_guard lk(_mutex);
            return _durable_msgs.size();
        }

        void clear()
        {
            std::lock_guard lk(_mutex);
            _helper.RemoveMessageFile();
            _ToBePushed_msgList.clear();
            _durable_msgs.clear();
            _waitConfirm_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }
        //
    private:
        bool GarbageCollectorCheck()
        {
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }
            return false;
        }
        void GarbageCollector()
        {
            if (false == GarbageCollectorCheck())
                return;
            std::list<MessagePtr> msgs = _helper.GarbageCollector();
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (_durable_msgs.end() == it)
                {
                    loge("有持久化消息，没有在内存管理");
                    _ToBePushed_msgList.push_back(msg);
                    _durable_msgs[msg->payload().properties().id()] = msg;
                    continue;
                }
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            _valid_count = _total_count = msgs.size();
        }
        bool InsertWithLock(MessagePtr msg)
        {
            {
                std::lock_guard lk(_mutex);
                if (msg->payload().properties().delivery_mode() == MQ_Protobuf::DeliveryMode::DURABLE)
                {
                    msg->mutable_payload()->set_valid(std::string("1"));
                    if (false == _helper.insert(msg))
                    {
                        loge("Failed to durably storage msg: {}", msg->mutable_payload()->body());
                        return false;
                    }
                    _valid_count += 1;
                    _total_count += 1;
                    _durable_msgs[msg->payload().properties().id()] = msg;
                }
                _ToBePushed_msgList.push_back(msg);
            }
            return true;
        }
        std::mutex _mutex;
        std::string _DataFilePath;
        size_t _valid_count;
        size_t _total_count;
        MessageHelper _helper;
        std::list<MessagePtr> _ToBePushed_msgList;
        std::unordered_map<std::string, MessagePtr> _durable_msgs;
        std::unordered_map<std::string, MessagePtr> _waitConfirm_msgs;
    };

    class MessageQueueManager
    {
    public:
        using self_ptr = std::shared_ptr<MessageQueueManager>;
        MessageQueueManager(const std::string baseDir)
            : _basedir(baseDir)
        {
            if (std::filesystem::is_directory(_basedir))
            {
                std::filesystem::create_directories(_basedir);
            }
            else
            {
                loge("the BaseDir Is not directory");
            }
        }
        void clear()
        {
            std::lock_guard lk(_mutex);
            for (auto &qmsg : _queue_messages)
            {
                qmsg.second->clear();
            }
        }
        void initQueueMessage(const std::string &qname)
        {
            MessageQueue::self_ptr qmp = nullptr;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() != it)
                {
                    return;
                }
                qmp = std::make_shared<MessageQueue>(GetFullPath(_basedir, qname).string());
                _queue_messages[qname] = qmp;
            }
            if (nullptr != qmp)
            {
                qmp->recovery();
            }
            else
            {
                logw("加载指针出错");
            }
            return;
        }
        void destroyQueueMessage(const std::string &qname)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    return;
                }
                qmp = it->second;
                _queue_messages.erase(it);
            }
            qmp->clear();
        }
        bool insert(const std::string &qname, const MQ_Protobuf::BasicProperties bp, const std::string &Body, bool queueIsDurable)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    loge("向队列 {}新增消息失败，没找到管理写出句柄", qname);
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, Body, queueIsDurable);
        }
        bool insert(const std::string &qname, const std::string &Body, bool queueIsDurable)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    logw("向队列 {}新增消息失败，没找到管理写出句柄", qname);
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(Body, queueIsDurable);
        }
        MessagePtr front(const std::string &qname)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    logw("从队列 {}获取消息失败，没找到管理写出句柄", qname);
                    return nullptr;
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        void Confirm(const std::string qname, const std::string &msg_id)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    logw("从队列 {}获取消息失败，没找到管理写出句柄", qname);
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
        }
        size_t ToBePushedMsg_Count(const std::string qname)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    logw("从队列 {}获取消息失败，没找到管理写出句柄", qname);
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getToBePushedMsg_count();
        }
        size_t Total_count(const std::string &qname)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    logw("从队列 {}获取消息失败，没找到管理写出句柄", qname);
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }

        size_t durable_count(const std::string &qname)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    loge("从队列 {}获取消息失败，没找到管理写出句柄", qname);
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        size_t waitConfirm_count(std::string &qname)
        {
            MessageQueue::self_ptr qmp;
            {
                std::lock_guard lk(_mutex);
                auto it = _queue_messages.find(qname);
                if (_queue_messages.end() == it)
                {
                    loge("从队列 {}获取消息失败，没找到管理写出句柄", qname);
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitConfirm_count();
        }

    private:
        std::filesystem::path GetFullPath(const std::string &BaseDirPath, const std::string &File_name)
        {
            std::filesystem ::path baseDirPath(BaseDirPath);
            std::filesystem ::path file_name(File_name);
            return baseDirPath / file_name;
        }
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, MessageQueue::self_ptr> _queue_messages;
    };
};